//******************************************************************
//******************************************************************
//**********          ANts Peer To Peer Sources        *************
//
// ANts P2P realizes a third generation P2P net. It protects your
// privacy while you are connected and makes you not trackable, hiding
// your identity (ip) and crypting everything you are sending/receiving
// from others.

// Copyright (C) 2004  Roberto Rossi

// This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU General Public License
// as published by the Free Software Foundation; either version 2
// of the License, or (at your option) any later version.

// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.

package ants.p2p.filesharing;

import java.io.*;
import java.util.*;
import ants.p2p.messages.*;
import ants.p2p.query.*;
import ants.p2p.utils.indexer.*;
import ants.p2p.utils.encoding.*;
import ants.p2p.*;
import kerjodando.casper.util.ConstantUtil;

import org.apache.log4j.*;

public class FilePullMessageProcessor extends Thread {
	PullWatchDog watchDog = new PullWatchDog(this);

	static Logger _logger = Logger.getLogger(FilePullMessageProcessor.class
			.getName());

	WarriorAnt n;

	FilePullMessage fpm;

	int percentage = 0;

	long startTime = System.currentTimeMillis();

	boolean terminate = false;

	public FilePullMessageProcessor(WarriorAnt n, FilePullMessage fpm) {
		this.n = n;
		this.fpm = fpm;
		this.setPriority(6);
	}

	public FilePullMessage getFilePullMessage() {
		return this.fpm;
	}

	public void run() {
		FileInputStream fis = null;
		watchDog.start();
		try {
			ants.p2p.security.EndpointSecurityManager esm = n
					.getInputSecureConnectionManager(fpm.getSource());
			if (esm == null) {
				SecureConnectionErrorControlMessage fterrcm = new SecureConnectionErrorControlMessage(
						new Integer(0), null, "No secure connection");
				MessageWrapper wm = n.sendMessage(fterrcm, fpm.getSource(),
						false, false);
				_logger.debug(n.getShortId() + ": No secure connection with "
						+ fpm.getSource());
				while (n.myMessages.contains(wm) && !this.terminate) {
					Thread.sleep(1000);
				}
				if (n.failedMessages.contains(wm)) {
					n.failedMessages.remove(wm);
					_logger.debug(n.getShortId()
							+ ": Error in sending ControlMessage "
							+ fterrcm.getMessage());
				}
				watchDog.terminate();
				throw new Exception(
						"No secure connection avaiable with endpoint "
								+ fpm.getSource());
			}

			try {
				fpm.decrypt(esm.getCipherDec());
			} catch (Exception e) {
				SecureConnectionErrorControlMessage fterrcm = new SecureConnectionErrorControlMessage(
						new Integer(0), null, "No secure connection");
				MessageWrapper wm = n.sendMessage(fterrcm, fpm.getSource(),
						false, false);
				_logger.debug(n.getShortId() + ": No secure connection with "
						+ fpm.getSource());
				while (n.myMessages.contains(wm) && !this.terminate) {
					Thread.sleep(1000);
				}
				if (n.failedMessages.contains(wm)) {
					n.failedMessages.remove(wm);
					_logger.debug(n.getShortId()
							+ ": Error in sending ControlMessage "
							+ fterrcm.getMessage());
				}
				watchDog.terminate();
				throw new Exception(
						"No secure connection avaiable with endpoint "
								+ fpm.getSource(), e);
			}

			if (n.checkInServiceFilePullRequests(false, fpm)) {
				watchDog.terminate();
				return;
			}

			byte[] hash = fpm.getHash();
			Long offset = fpm.getOffset();
			File f = null;

			// controlla ultima modifica e caratteristiche del file
			BackgroundEngine be = BackgroundEngine.getInstance();
			boolean shared = false;
			boolean partial = false;
			String fpmHash = Base16.toHexString(hash);
			FileInfos fileInfo = (FileInfos) be.sharedFilesIndexHash
					.get(fpmHash);
			if (fileInfo != null) {
				shared = true;
				f = new File(fileInfo.getName());
			}
			if (!shared || f == null || !f.exists()) {
				Object partialFile = BackgroundEngine.getInstance()
						.getPartialFile(fpmHash, QueryHashItem.ANTS_HASH);
				if (partialFile != null) {
					shared = true;
					long partId = fpm.getOffset().longValue()
							/ (fpm.getBlockSize().intValue() * fpm.getBlocks()
									.longValue());
					f = new File(WarriorAnt.chunksHome + WarriorAnt.chunksPath
							+ fpmHash + "." + partId);
					partial = true;
				}
			}
			if (!shared || f == null || !f.exists()) {
				FileTransferErrorControlMessage fterrcm = new FileTransferErrorControlMessage(
						new Integer(0), hash, "File don't exist", fpm);
				fterrcm.encrypt(esm.getCipherEnc());
				MessageWrapper wm = n.sendMessage(fterrcm, fpm.getSource(),
						false, false);
				_logger.debug(n.getShortId() + ": File " + fpm.getFileName()
						+ " don't exist. Cannot serve pull request from id "
						+ fpm.getSource());
				while (n.myMessages.contains(wm) && !this.terminate) {
					Thread.sleep(1000);
				}
				if (n.failedMessages.contains(wm)) {
					n.failedMessages.remove(wm);
					fterrcm.decrypt(esm.getCipherDec());
					_logger.debug(n.getShortId()
							+ ": Error in sending ControlMessage "
							+ fterrcm.getMessage());
				}
				watchDog.terminate();
				return;
			}

			if (n.checkInServiceFilePullRequests(true, fpm)) {
				watchDog.terminate();
				return;
			}

			fis = new FileInputStream(f);
			FilePushMessage frm = new FilePushMessage(fpm.getLocalFileName(),
					hash, offset, fpm.getBlockSize(), fpm.getResume(),
					new Long(f.length()));
			frm.encrypt(esm.getCipherEnc());
			MessageWrapper wm = n.sendMessage(frm, fpm.getSource(), false,
					false);
			while (n.myMessages.contains(wm) && !n.isDisconnected()
					&& !this.terminate) {
				Thread.sleep(1000);
			}
			if (n.failedMessages.contains(wm)) {
				n.failedMessages.remove(wm);
				frm.decrypt(esm.getCipherDec());
				_logger.debug(n.getShortId()
						+ ": Error in sending FilePushMessage "
						+ frm.getFileName()
						+ " - Cannot serve pull request from id "
						+ fpm.getSource());
				n.inServiceFilePullRequests.remove(fpm);
				watchDog.terminate();
				return;
			}
			watchDog.setReceivedPart();

			frm.decrypt(esm.getCipherDec());
			n.propertyChangeSupport.firePropertyChange("filePullInit", null,
					this);

			byte[] block = new byte[fpm.getBlockSize().intValue()];
			long blocks = fpm.getBlocks().longValue();
			if (!partial) {
				fis.skip(offset.longValue());
			}
			long partId = 0;
			long partsUploaded = 0;
			ArrayList messageGroup = new ArrayList();
			long sizeUploaded = 0;
			while (fis.available() > 0 && partId < blocks && !this.terminate) {
				for (int g = 0; g < (PartialFile.computeGroupFactor(fpm
						.getBlockSize().intValue()) - messageGroup.size())
						&& partId < blocks && !this.terminate; g++) {
					int read = fis.read(block);
					if (read > 0) {
						if (read < block.length) {
							byte[] reduced = new byte[read];
							for (int x = 0; x < read; x++) {
								reduced[x] = block[x];
							}
							block = reduced;
						}
						FilePartMessage filepartmess = new FilePartMessage(
								block, hash, new Long(partId), frm);
						filepartmess.encrypt(esm.getCipherEnc());
						MessageWrapper wm1 = n.sendMessage(filepartmess, fpm
								.getSource(), false, false);
						sizeUploaded += block.length;
						partId++;
						messageGroup.add(wm1);
						while (messageGroup.size() >= 2
								&& !n.areInFailedMessages(messageGroup)
								&& !this.terminate) {
							int lastSize = messageGroup.size();
							partsUploaded += messageGroup.size();
							messageGroup = n
									.removeDeliveredMessagesFromList(messageGroup);
							int currentSize = messageGroup.size();
							partsUploaded -= messageGroup.size();
							if (lastSize > currentSize) {
								watchDog.setReceivedPart();
							}
							this.percentage = (int) (((partsUploaded * 1.0) / blocks) * 100.0);
							n.propertyChangeSupport.firePropertyChange(
									"filePullUpdate", null, this);
							Thread.sleep(1000);
						}
						if (n.areInFailedMessages(messageGroup)) {
							n.failedMessages.removeAll(messageGroup);

							n.inServiceFilePullRequests.remove(fpm);
							n.propertyChangeSupport.firePropertyChange(
									"filePullError", null, this
											.getFilePullMessage());
							_logger.info(n.getShortId()
									+ ": Error in file transfer "
									+ fpm.getFileName()
									+ ". Cannot serve pull request from id "
									+ fpm.getSource());
							fis.close();

							watchDog.terminate();
							return;
						}
					}
				}
			}
			while (messageGroup.size() > 0
					&& !n.areInFailedMessages(messageGroup) && !this.terminate) {
				int lastSize = messageGroup.size();
				partsUploaded += messageGroup.size();
				messageGroup = n.removeDeliveredMessagesFromList(messageGroup);
				int currentSize = messageGroup.size();
				partsUploaded -= messageGroup.size();
				if (lastSize > currentSize) {
					watchDog.setReceivedPart();
				}
				this.percentage = (int) (((partsUploaded * 1.0) / blocks) * 100.0);
				n.propertyChangeSupport.firePropertyChange("filePullUpdate",
						null, this);
				Thread.sleep(1000);
			}
			if (n.areInFailedMessages(messageGroup)) {
				n.failedMessages.removeAll(messageGroup);

				n.inServiceFilePullRequests.remove(fpm);
				n.propertyChangeSupport.firePropertyChange("filePullError",
						null, this.getFilePullMessage());
				_logger.info(n.getShortId() + ": Error in file transfer "
						+ fpm.getFileName()
						+ ". Cannot serve pull request from id "
						+ fpm.getSource());
				fis.close();

				watchDog.terminate();
				return;
			}

			fis.close();

			n.inServiceFilePullRequests.remove(fpm);
			FileTransferEndControlMessage cm = new FileTransferEndControlMessage(
					new Integer(0), hash, "", frm);
			cm.encrypt(esm.getCipherEnc());
			MessageWrapper wm2 = n.sendMessage(cm, fpm.getSource(), false,
					false);
			while (n.myMessages.contains(wm2) && !this.terminate) {
				Thread.sleep(1000);
			}
			if (n.failedMessages.contains(wm2)) {
				n.failedMessages.remove(wm2);

				n.propertyChangeSupport.firePropertyChange("filePullError",
						null, this.getFilePullMessage());
				_logger.info(n.getShortId() + ": Error in file transfer "
						+ fpm.getFileName()
						+ ". Cannot serve pull request from id "
						+ fpm.getSource());
				watchDog.terminate();
				return;
			}
			watchDog.setReceivedPart();

			NeighbourAnt.totalUploaded += sizeUploaded;
			n.propertyChangeSupport.firePropertyChange("filePullEnd", null,
					this.getFilePullMessage());
			_logger.info(n.getShortId() + ": File sent " + fpm.getFileName()
					+ " - " + fpm.getLocalFileName()
					+ ". Pull request from id " + fpm.getSource() + " served.");
			watchDog.terminate();
		} catch (Exception e) {
			try {
				fis.close();
			} catch (Exception ex) {
			}

			n.inServiceFilePullRequests.remove(fpm);
			n.propertyChangeSupport.firePropertyChange("filePullError", null,
					this.getFilePullMessage());
			watchDog.terminate();

			_logger.error(n.getShortId()
					+ ": Error In Processing FilePullMessage: id = "
					+ fpm.getAck_Id() + " Source: " + fpm.getSource()
					+ " Dest: " + fpm.getDest(), e);
		}
	}

	public void terminate() {
		this.terminate = true;
		this.watchDog.terminate();
		try {
			ants.p2p.security.EndpointSecurityManager esm = n
					.getInputSecureConnectionManager(fpm.getSource());
			if (esm == null) {
				SecureConnectionErrorControlMessage fterrcm = new SecureConnectionErrorControlMessage(
						new Integer(0), null, "No secure connection");
				MessageWrapper wm = n.sendMessage(fterrcm, fpm.getSource(),
						false, false);
				_logger.debug(n.getShortId() + ": No secure connection with "
						+ fpm.getSource());
				while (n.myMessages.contains(wm) && !this.terminate) {
					Thread.sleep(1000);
				}
				if (n.failedMessages.contains(wm)) {
					n.failedMessages.remove(wm);
					_logger.debug(n.getShortId()
							+ ": Error in sending ControlMessage "
							+ fterrcm.getMessage());
				}
				watchDog.terminate();
				_logger.error("No secure connection avaiable with endpoint "
						+ fpm.getSource());
				return;
			}

			FileTransferErrorControlMessage cm = new FileTransferErrorControlMessage(
					new Integer(1), fpm.getHash(), "", fpm);
			cm.encrypt(esm.getCipherEnc());
			MessageWrapper wm2 = n.sendMessage(cm, fpm.getSource(), false,
					false);
			while (n.myMessages.contains(wm2)) {
				Thread.sleep(1000);
			}
			if (n.failedMessages.contains(wm2)) {
				n.failedMessages.remove(wm2);
				_logger.info(n.getShortId()
						+ ": Error file transfer interruption "
						+ fpm.getFileName()
						+ ". Cannot notificate interruption of msg id "
						+ fpm.getSource());
				return;
			}
			_logger.warn(n.getShortId() + ": File pull interrupted: id = "
					+ fpm.getAck_Id() + " Source: " + fpm.getSource()
					+ " Dest: " + fpm.getDest());
		} catch (Exception ex) {
			_logger
					.error(
							"Exception while trying to notify file transfer interruption",
							ex);
		}
	}

	public double getSpeedValue() {
		double elapsed = (System.currentTimeMillis() - this.startTime) / 1000.0;
		double speed = (this.percentage / 100.0)
				* (fpm.getBlocks().longValue() * fpm.getBlockSize().longValue())
				/ (elapsed * ConstantUtil.KBYTES);
		return speed > 0 ? speed : 0;
	}

	public String speedToString() {
		double speed = this.getSpeedValue();
		return ("" + speed).substring(0, ("" + speed).indexOf(".") + 2)
				+ "KB/s ";
	}

	public String toString() {
		String rv = percentage + "%";
		while (rv.length() < 7) {
			rv += " ";
		}
		rv += this.speedToString();
		while (rv.length() < 15) {
			rv += " ";
		}
		rv += "     " + ji.JI.i("Hash") + ": " + fpm.getLocalFileName();
		while (rv.length() < 80) {
			rv += " ";
		}
		if (this.n.acceptTCPDirectConnections()
				&& !fpm.getSourceAddress().equals("")) {
			rv += ji.JI.i("Requirer") + ": " + fpm.getSourceAddress();
		} else {
			rv += ji.JI.i("Requirer") + ": " + fpm.getSource().substring(0, 10);
		}
		rv += "    " + ji.JI.i("Name") + ": " + fpm.getFileName();
		return rv;
	}
}

class PullWatchDog extends Thread {
	public static long downloadTimeout = Ant.messageTimeout
			* Ant.maxRetransmissions;

	long lastPartReceivedTime;

	FilePullMessageProcessor fpmp = null;

	boolean interrupt = false;

	static Logger _logger = Logger.getLogger(PullWatchDog.class.getName());

	public PullWatchDog(FilePullMessageProcessor fpmp) {
		super();
		this.fpmp = fpmp;
		lastPartReceivedTime = System.currentTimeMillis();
		this.setPriority(1);
	}

	public void setReceivedPart() {
		lastPartReceivedTime = System.currentTimeMillis();
	}

	public void terminate() {
		this.interrupt = true;
	}

	public void run() {
		while (fpmp.isAlive() && !interrupt) {
			try {
				sleep(5000);
			} catch (Exception e) {
				_logger.error("", e);
			}
			long elapsed = System.currentTimeMillis()
					- this.lastPartReceivedTime;
			if (elapsed > downloadTimeout || fpmp.n.isDisconnected()) {
				fpmp.terminate();
				_logger.info("File pull message processor timed out");
				fpmp.n.propertyChangeSupport.firePropertyChange(
						"filePullError", null, fpmp.getFilePullMessage());
				fpmp.n.inServiceFilePullRequests.remove(fpmp.fpm);
				return;
			}
		}
	}
}
